package comsumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * @author 夏焰波
 * @date 2020-09-13 23:53
 * <p>
 * KafkaConsumer： 需要创建一个消费者对象，用来消费数据
 * ConsumerConfig： 获取所需的一系列配置参数
 * ConsumerRecord： 每条数据都要封装成一个 ConsumerRecord对象
 * 为了使我们能够专注于自己的业务逻辑，Kafka提供了自动提交 offset的功能 。
 * 自动提交 offset的相关参数：
 * enable.auto.commit： 是否开启自动提交 offset功能
 * auto.commit.interval.ms： 自动提交 offset的时间间隔
 */
public class MyConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        //连接的集群
        props.put("bootstrap.servers", "hadoop102:9092");
        //自动提交offset的代码
        props.put("enable.auto.commit", "true");
        //自动提交的延时
        props.put("auto.commit.interval.ms", "1000");

        //Key，Value的反序列化
        props.put("key.deserializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.deserializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        //消费者组
        props.put("group.id", "test");
        //重置消费者的offset，earliest、latest(默认)
        // 触发条件：换了组或消费的数据已经不存在
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");

        //创建消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        //订阅主题
        consumer.subscribe(Arrays.asList("first", "second"));
        while (true) {
            //获取数据
            //ConsumerRecords<String, String> records = consumer.poll(100);
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            //解析并打印consumerRecords
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
        }
        //关闭连接，前边有无限循环while
        // 如果开启了自动提交，默认30秒关闭连接
        //consumer.close();
    }
}
